package com.data.mall.service.impl;

import com.data.mall.common.KafkaTrackingDTO;
import com.data.mall.domain.Item;
import com.data.mall.domain.ItemStock;
import com.data.mall.domain.StockLog;
import com.data.mall.etl.domain.enums.Event;
import com.data.mall.etl.domain.enums.Params;
import com.data.mall.exception.BusinessException;
import com.data.mall.kafka.KafkaProduceService;
import com.data.mall.repositry.ItemRepository;
import com.data.mall.repositry.ItemStockRepository;
import com.data.mall.repositry.StockLogRepository;
import com.data.mall.service.ItemService;
import com.data.mall.service.PromoService;
import com.data.mall.service.dto.ItemDTO;
import com.data.mall.service.dto.PromoDTO;
import com.data.mall.utils.JSON;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.LocalDateTime;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @author shixukai
 * @Package com.data.mall.service.impl
 * @Description: TODO
 * @date 2021/4/11
 */
@Service
public class ItemServiceImpl implements ItemService {

    @Autowired
    private ItemRepository itemRepository;

    @Autowired
    private PromoService promoService;

    @Autowired
    private ItemStockRepository itemStockRepository;

    @Autowired
    private StockLogRepository stockLogRepository;

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    KafkaProduceService kafkaProduceService;

    private Item convertItemDOFromItemModel(ItemDTO itemDTO) {
        if (itemDTO == null) {
            return null;
        }
        Item item = new Item();
        BeanUtils.copyProperties(itemDTO, item);
        item.setPrice(Double.valueOf(itemDTO.getPrice()));
        return item;
    }

    private ItemStock convertItemStockDOFromItemModel(ItemDTO itemModel) {
        if (itemModel == null) {
            return null;
        }
        ItemStock itemStockDO = new ItemStock();
        itemStockDO.setItemId(itemModel.getId());
        itemStockDO.setStock(itemModel.getStock());
        return itemStockDO;
    }

    @Override
    @Transactional
    public ItemDTO createItem(ItemDTO itemDTO) throws BusinessException {
        //校验入参
        //转化itemmodel->dataobject
        Item itemDO = this.convertItemDOFromItemModel(itemDTO);

        //写入数据库
        itemRepository.save(itemDO);
        itemDTO.setId(itemDO.getId());

        ItemStock itemStockDO = this.convertItemStockDOFromItemModel(itemDTO);
        itemStockRepository.save(itemStockDO);
        ItemDTO res = this.getItemById(itemDTO.getId());
        // 埋点
        KafkaTrackingDTO kafkaTrackingDTO = KafkaTrackingDTO.builder()
                .event(Event.CREATE_ITEM)
                .param(Params.ITEM, JSON.toJSONString(res))
                .build();
        kafkaProduceService.sendTrackingMessage(kafkaTrackingDTO);
        //返回创建完成的对象
        return res;
    }

    @Override
    public List<ItemDTO> listItem() {
        List<Item> itemDOList = itemRepository.findAll();
        List<ItemDTO> itemModelList = itemDOList.stream().map(itemDO -> {
            ItemStock itemStockDO = itemStockRepository.findByItemId(itemDO.getId());
            ItemDTO itemModel = this.convertModelFromDataObject(itemDO, itemStockDO);
            return itemModel;
        }).collect(Collectors.toList());
        return itemModelList;
    }

    @Override
    public ItemDTO getItemById(Long id) {
        Item itemDO = itemRepository.findById(id).orElse(null);
        if (itemDO == null) {
            return null;
        }
        //操作获得库存数量
        ItemStock itemStockDO = itemStockRepository.findByItemId(itemDO.getId());
        //将dataobject->model
        ItemDTO itemModel = convertModelFromDataObject(itemDO, itemStockDO);

        //获取活动商品信息
        PromoDTO promoModel = promoService.getPromoByItemId(itemModel.getId());
        if (promoModel != null && promoModel.getStatus().intValue() != 3) {
            itemModel.setPromoDTO(promoModel);
        }
        return itemModel;
    }

    @Override
    public ItemDTO getItemByIdInCache(Long id) {
        ItemDTO itemModel = (ItemDTO) redisTemplate.opsForValue().get("item_validate_" + id);
        if (itemModel == null) {
            itemModel = this.getItemById(id);
            redisTemplate.opsForValue().set("item_validate_" + id, itemModel);
            redisTemplate.expire("item_validate_" + id, 10, TimeUnit.MINUTES);
        }
        return itemModel;
    }

    @Override
    @Transactional
    public boolean decreaseStock(Long itemId, Integer amount) throws BusinessException {
        long result = redisTemplate.opsForValue().increment("promo_item_stock_" + itemId, amount.intValue() * -1);
        if (result > 0) {
            //更新库存成功
            return true;
        } else if (result == 0) {
            //打上库存已售罄的标识
            redisTemplate.opsForValue().set("promo_item_stock_invalid_" + itemId, "true");

            //更新库存成功
            return true;
        } else {
            //更新库存失败
            increaseStock(itemId, amount);
            return false;
        }
    }

    @Override
    public boolean increaseStock(Long itemId, Integer amount) throws BusinessException {
        redisTemplate.opsForValue().increment("promo_item_stock_" + itemId, amount.intValue());
        return true;
    }

    @Override
    public void increaseSales(Long itemId, Integer amount) throws BusinessException {
        itemRepository.increaseSales(itemId, amount);
    }

    //初始化对应的库存流水
    @Override
    @Transactional
    public String initStockLog(Long itemId, Integer amount) {
        StockLog stockLogDO = new StockLog();
        stockLogDO.setItemId(itemId);
        stockLogDO.setAmount(amount);
        stockLogDO.setStockLogId(UUID.randomUUID().toString().replace("-", ""));
        stockLogDO.setStatus(1);

        stockLogRepository.save(stockLogDO);

        return stockLogDO.getStockLogId();
    }

    private ItemDTO convertModelFromDataObject(Item itemDO, ItemStock itemStockDO) {
        ItemDTO itemModel = new ItemDTO();
        BeanUtils.copyProperties(itemDO, itemModel);
        itemModel.setPrice(itemDO.getPrice().toString());
        itemModel.setStock(itemStockDO.getStock());
        return itemModel;
    }
}
